package com.xsoft.sms.cmpp20.comm.impl;

import com.xsoft.sms.cmpp20.comm.ICMPP20Comm;
import com.xsoft.sms.cmpp20.comm.listener.IClientListener;
import com.xsoft.sms.cmpp20.config.CMPP20Config;
import com.xsoft.sms.cmpp20.message.factory.CMPP20MessageFactory;
import com.xsoft.sms.cmpp20.message.impl.CMPP20Message;
import com.xsoft.sms.cmpp20.message.impl.CMPP20MessageConnectReqBody;
import com.xsoft.sms.cmpp20.message.impl.CMPP20MessageHead;
import com.xsoft.sms.cmpp20.msg.SmsMsg;
import org.apache.log4j.Logger;

import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * package: {PACKAGE_NAME}
 * User: zhanglm
 * Date: 12-9-18
 * Time: 上午11:21
 * Description:
 */
public final class CMPP20Comm implements ICMPP20Comm, IClientListener {

	private static Logger logger = Logger.getLogger(CMPP20Comm.class);


	private final AtomicInteger sequenceId = new AtomicInteger(0);
	private final AtomicInteger sendingCount = new AtomicInteger(0);

	private final CMPP20SendTask sender;
	private final CMPP20ResendTask reSender;
	private final CMPP20FillSendQueueTask fillSendQueueTask;
	private Timer resendTimer = null;

	private final Object fillContition = new Object();

	private final ConcurrentLinkedQueue<SmsMsg> sendingQueue = new ConcurrentLinkedQueue<SmsMsg>();
	private final ConcurrentLinkedQueue<SmsMsg> waitingQueue = new ConcurrentLinkedQueue<SmsMsg>();
	//	private final ConcurrentLinkedQueue<SmsMsg> resendQueue = new ConcurrentLinkedQueue<SmsMsg>();
	private final ConcurrentHashMap<Integer, SmsMsg> resendQueue = new ConcurrentHashMap<Integer, SmsMsg>();

	private boolean running = false;
	private final CMPP20Client client;

	public CMPP20Comm() {
		try {
			Class.forName("com.xsoft.sms.cmpp20.config.CMPP20Config", true, CMPP20Config.class.getClassLoader());
		} catch (ClassNotFoundException ex) {
			ex.printStackTrace();
			logger.fatal(ex.getMessage());
		}
		/**
		 * 初始化网络连接
		 */
		client = new CMPP20Client(CMPP20Config.getGwAddr(), CMPP20Config.getGwPort(), this);
		/**
		 * 初始化任务
		 */
		fillSendQueueTask = new CMPP20FillSendQueueTask(CMPP20Config.getWindowSize());
		sender = new CMPP20SendTask();
		reSender = new CMPP20ResendTask(CMPP20Config.getResendInterval(), CMPP20Config.getResendCount());
	}

	@Override
	public void start() {
		boolean connResult = client.connect();
		if (connResult) {
			cmppConnect();
		}
	}

	@Override
	public void stop() {
		client.disconnect();
	}


	@Override
	public void idleTimeout(Long seconds) {
		//To change body of implemented methods use File | Settings | File Templates.
	}

	@Override
	public void clientDisconnected() {
		client.disconnect();
	}

	@Override
	public void clientConnected() {
		//连接上之后立即发送登陆包
		cmppConnect();
	}

	@Override
	public void cmppSubmit(String[] dest, byte[] content, byte msgFmt) {
		int newSeqId = getNextSequenceId();
		SmsMsg msg = new SmsMsg(newSeqId, content, dest, msgFmt);
		synchronized (fillContition) {
			waitingQueue.add(msg);
			fillContition.notifyAll();
		}
	}

	@Override
	public void cmppSubmit(String[] dest, String content) {
		byte[] contentData = content.getBytes(Charset.forName("gb2312"));
		cmppSubmit(dest, contentData, CMPP20Message.CMPP20_DEFAULT_MSG_GBHZ);
	}


	@Override
	public void messageReceived(CMPP20Message message) {
		CMPP20MessageHead head = (CMPP20MessageHead) message.getHead();
		int cmdId = head.getCmdId();
		switch (cmdId) {
			case CMPP20Message.CMPP20_CONNECT_RESP:
				onConnectResp(message);
				break;
			case CMPP20Message.CMPP20_SUBMIT_RESP:
				onSubmitResp(message);
				break;
			case CMPP20Message.CMPP20_ACTIVE_TEST:
				onActiveTest(message);
				break;
			case CMPP20Message.CMPP20_ACTIVE_TEST_RESP:
				onActiveTestResp(message);
				break;
			case CMPP20Message.CMPP20_DELIVER:
				onDeliver(message);
				break;
			default:
				logger.info("未知命令码:0x" + Integer.toHexString(cmdId).toUpperCase());
				break;
		}
	}

	private void cmppConnect() {
		CMPP20Message message = CMPP20MessageFactory.newCmpp20Message(CMPP20Message.CMPP20_CONNECT, getNextSequenceId());
		CMPP20MessageConnectReqBody reqBody = (CMPP20MessageConnectReqBody) message.getBody();
		reqBody.setSpId(CMPP20Config.getSpId());
		reqBody.setPassword(CMPP20Config.getPassword());
		client.send(message);
	}

	private void onActiveTestResp(CMPP20Message message) {
		//To change body of created methods use File | Settings | File Templates.
	}

	private void onDeliver(CMPP20Message message) {
		//To change body of created methods use File | Settings | File Templates.
	}

	private void onActiveTest(CMPP20Message message) {
		//To change body of created methods use File | Settings | File Templates.
	}

	private void onSubmitResp(CMPP20Message message) {
		CMPP20MessageHead head = (CMPP20MessageHead) message.getHead();
		int seqId = head.getSequenceId();
		if(null != resendQueue.remove(seqId)) {
			synchronized (fillContition) {
				sendingCount.decrementAndGet();
				fillContition.notifyAll();
			}
		}
	}

	private void onConnectResp(CMPP20Message message) {

	}


	private int getNextSequenceId() {
		return sequenceId.getAndDecrement();
	}



	private void fullFillSendingQueue(int count) {
		synchronized (sendingQueue) {
			for (int i = 0; i < count; i++) {
				SmsMsg msg = waitingQueue.poll();
				if (msg == null) {
					break;
				}
				sendingQueue.add(msg);
			}
			logger.debug("after fullFillSendingQueue, sendingQueue:" + sendingQueue.size());
			sendingQueue.notifyAll();
		}
	}


	private void fillResendQueue(SmsMsg msg) {
		msg.updateTimestamp();
		msg.updateSendCount();
		resendQueue.put(msg.getSeqId(), msg);
	}



	private class CMPP20SendTask implements Runnable {
		@Override
		public void run() {
			while (running) {
				synchronized (sendingQueue) {
					while (running && sendingQueue.isEmpty()) {
						try {
							sendingQueue.wait(2 * 1000);
						} catch (InterruptedException e) {
							logger.error(e.getMessage());
							e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
						}
					}
				}
				if (running && !sendingQueue.isEmpty()) {
					SmsMsg msg;
					while ((msg = sendingQueue.poll()) != null) {
						fillResendQueue(msg);
					}
				}
			}
			logger.info("发送线程退出");
		}
	}

	private class CMPP20ResendTask extends TimerTask {
		private final int resendTime;
		private final int retryCount;

		private CMPP20ResendTask(int resendTime, int retryCount) {
			this.resendTime = resendTime;
			this.retryCount = retryCount;
		}

		@Override
		public void run() {
			long currentTimestamp = new Date().getTime() / 1000;
			Iterator iterator = resendQueue.entrySet().iterator();
			while (iterator.hasNext()) {
				Map.Entry<Long, SmsMsg>  entry = (Map.Entry<Long, SmsMsg>) iterator.next();
				SmsMsg msg = entry.getValue();
				long seqId = entry.getKey();
				long msgTimestamp = msg.getTimestamp();
				if (currentTimestamp > msgTimestamp &&
						(currentTimestamp - msgTimestamp) > resendTime) {

					if (msg.getSendCount() >= retryCount) {
						iterator.remove();
					}
				}

			}
		}
	}

	private class CMPP20FillSendQueueTask implements Runnable {
		private final int windowSize;

		private CMPP20FillSendQueueTask(int windowSize) {
			this.windowSize = windowSize;
		}

		@Override
		public void run() {
			while(running) {
				synchronized (fillContition) {
					logger.debug("sendingCount:" + sendingCount.get() + " waitingQueue:" + waitingQueue.size());
					if (sendingCount.get() >= windowSize || waitingQueue.isEmpty()) {
						try {
							fillContition.wait(2 * 1000);
						} catch (InterruptedException e) {
							logger.error(e.getMessage());
							e.printStackTrace();
						}
					}
					if (!waitingQueue.isEmpty()) {
						int count = windowSize - sendingCount.get();
						fullFillSendingQueue(count);
						logger.debug("滑动窗口空余:" + count);
					}
				}
			}
			logger.info("滑动窗口线程退出");
		}
	}
}
